package com.l.im.common.mq;


import java.io.IOException;
import java.util.Collections;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

/**
 * @program: LIM
 * @description: 信息消费类
 * @author: Alex Wu
 * @createDate: 2025-04-15 14:28
 **/
@Slf4j
public class RocketMQConsumer {

    private PushConsumer pushConsumer;

    public void start(String endpoint, String topic, String group, String tag, MessageListener handler) throws ClientException {
        ClientServiceProvider provider = ClientServiceProvider.loadService();

        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint)
                .build();

        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(group)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(handler)
                .build();

        log.info("RocketMQ consumer started: topic={}, group={}", topic, group);
    }

    public void shutdown() {
        if (pushConsumer != null) {
            try {
                pushConsumer.close();
                log.info("RocketMQ consumer shutdown");
            } catch (IOException e) {
                log.error("Error shutting down RocketMQ consumer", e);
            }

        }
    }
}
